package com.atbeijing.D09;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 分流求平均温度
 * table api 方式
 * sql 方式
 */
public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //表环境配置
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        //获取表环境
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env, settings);

        //数据流源
        DataStreamSource<SensorReading> stream = env.addSource(new SensorSource());


        //table api
        //pojo字段名与$("id")相匹配
        //1.DataStream => Dynamic Table; 数据流 => 动态表  as：别名
        Table table = tableEnvironment.fromDataStream(stream, $("id"), $("temperature").as("temp"), $("timestamp").as("ts"));
        //2.在动态表上进行查询, 查询结果也是动态表
        Table tableResult = table
                .groupBy($("id"))
                .select($("id"), $("temp").avg());
        //3.将动态表转换成数据流：Dynamic Table => DataStream
        //toRetractStream撤回流,有聚合情况下使用,false表示数据失效撤回,true表示正确
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnvironment.toRetractStream(tableResult, Row.class);
        tuple2DataStream.print();

        //sql方式
        // 1.将数据流转换成临时表
        tableEnvironment.createTemporaryView("sensor", stream, $("id"), $("temperature").as("temp"), $("timestamp").as("ts"));
        // 2.在临时表上进行sql查询
        Table sqlResult = tableEnvironment
                .sqlQuery("SELECT id, AVG(temp) FROM sensor GROUP BY id");
        // 3.将动态表转换成数据流
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream1 = tableEnvironment.toRetractStream(sqlResult, Row.class);
        tuple2DataStream1.print();


        env.execute();
    }
}
